package com.demo.rpc.zk;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

public class ZkClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(ZkClient.class);
    private int timeout;
    private String zkServers;
    private String serverPath;
    private ZooKeeper zk;
    private Watcher watcher;
    private volatile List<String> serverList;


    public ZkClient() {
        super();
    }

    public ZkClient(String zkServers, String serverPath) {
        this(zkServers, serverPath, 20000);
    }

    public ZkClient(String zkServers, String serverPath, int timeout) {
        super();
        this.timeout = timeout;
        this.zkServers = zkServers;
        this.serverPath = serverPath;
        init();
    }


    public void init() {
        watcher = new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getType() == Event.EventType.NodeChildrenChanged && serverPath.equals(event.getPath())) {
                    updateServerList();
                }
            }
        };
        getZk();
    }

    private void getZk() {
        try {
            zk = new ZooKeeper(zkServers, timeout, watcher);
            Stat stat = zk.exists(serverPath, watcher);
            if (stat == null) {
                throw new RuntimeException("服务器未启动");
            } else {
                updateServerList();
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    protected void updateServerList() {
        try {
            List<String> servers = zk.getChildren(serverPath, watcher);
            List<String> server1 = new ArrayList<>();
            for (String server : servers) {
                byte[] data = zk.getData(serverPath + "/" + server, false, null);
                server1.add(new String(data, StandardCharsets.UTF_8));
            }
            serverList = server1;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void closeZk(){
        if(zk!=null){
            try {
                zk.close();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public synchronized String getServer(){
        String server = null;
        if(serverList.size()>0){
            server=serverList.remove(0);
            serverList.add(server);
            LOGGER.debug("get server:"+server);
        }
        return server;
    }

    public List<String> getServerAll(){
        return serverList;
    }
}
